package ItemBased;

import java.io.IOException;
import java.util.Iterator;
import java.util.Vector;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

public class GenerateUserVector {

	
	public static class GenerateUserVectorMap
    extends Mapper<LongWritable, LongPairArrayWritableComparable, LongWritable, LongPairWritableComparable>{
	      
	    public void map(LongWritable key, LongPairArrayWritableComparable value,
	    		OutputCollector<LongWritable, LongPairWritableComparable> output, Reporter reporter
	                    ) throws IOException, InterruptedException {
	    	//normalize here
	    	
	    	for(LongPairWritableComparable p : value.getData()){
	    		output.collect(new LongWritable(p.getFirst()), new LongPairWritableComparable(key, new LongWritable(p.getSecond())));
	    	}
	    	
	    }
		
	}	
	
	
	public static class GenerateUserVectorReduce
    extends Reducer<Text,IntWritable,Text,IntWritable> {
		
		public void reduce(LongWritable key, Iterator<LongPairWritableComparable> values,
	    		OutputCollector<LongWritable, LongPairArrayWritableComparable> output, Reporter reporter
	                    ) throws IOException, InterruptedException {
			Vector<LongPairWritableComparable> list = new Vector<LongPairWritableComparable> (); 
			while(values.hasNext()){
				list.add(values.next().get());
			}
			
			LongPairWritableComparable[] pair = new LongPairWritableComparable[list.size()];
			LongPairArrayWritableComparable arrayPair = new LongPairArrayWritableComparable();
			
			for(int i=0; i<list.size();i++){
				pair[i] = list.get(i);
			}
			arrayPair.setData(pair);
			output.collect(key, arrayPair);
		}
	}	
}
